package com.suppor.rpc.proto.codec;


import com.suppor.rpc.core.trans.RpcRequest;
import com.suppor.rpc.core.trans.RpcResponse;
import com.suppor.rpc.proto.protocol.MsgHeader;
import com.suppor.rpc.proto.protocol.MsgType;
import com.suppor.rpc.proto.protocol.RpcProtocol;
import com.suppor.rpc.proto.serialization.RpcSerializer;
import com.suppor.rpc.proto.serialization.SerializationFactory;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;

public class RpcDecoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> out) throws Exception {
        if (in.readableBytes() < MsgHeader.HEAD_LENGTH) {
            //判断是否读到了头部的长度的数据
            return;
        }
        in.markReaderIndex();
        MsgHeader header = new MsgHeader();
        header.setMagic(in.readShort());
        header.setVersion(in.readByte());
        header.setSerializeType(in.readByte());
        header.setMsgType(in.readByte());
        header.setStatus(in.readByte());
        header.setRequestId(in.readLong());

        int dataLeLen = in.readInt();
        if (in.readableBytes() < dataLeLen) {
            // 不是完整的数据报
            in.resetReaderIndex();
            return;
        }
        byte[] data = new byte[dataLeLen];
        in.readBytes(data);

        RpcSerializer serializer = SerializationFactory.get(header.getSerializeType());
        MsgType msgType = MsgType.findType(header.getMsgType());
        switch (msgType) {
            case REQUEST:
                RpcRequest request = serializer.deserialize(data, RpcRequest.class);
                RpcProtocol<RpcRequest> protocol = new RpcProtocol<>();
                protocol.setHeader(header);
                protocol.setBody(request);
                out.add(protocol);
                break;
            case RESPONSE:
                RpcResponse response = serializer.deserialize(data, RpcResponse.class);
                RpcProtocol<RpcResponse> responseRpcProtocol = new RpcProtocol<>();
                responseRpcProtocol.setHeader(header);
                responseRpcProtocol.setBody(response);
                out.add(responseRpcProtocol);
                break;
            case HEARTBEAT:
                header.setLength(0);
                RpcProtocol<RpcRequest> heart = new RpcProtocol<>();
                heart.setHeader(header);
                heart.setBody(null);
                out.add(heart);
                break;
            default:
                throw new RuntimeException("Not Support Type: " + msgType.getType());

        }
    }
}
